package com.maozzi.api.mq.template;

import com.google.gson.Gson;
import com.maozzi.api.mq.constant.MqConstant;
import com.maozzi.api.mq.properties.MqProperties;
import com.maozzi.api.mq.model.BaseMsg;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;

/**
 * message 模版
 *
 * @author maozi
 */
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class MsgTemplate {
    @Getter
    private final RocketMQTemplate template;

    @Resource
    private MqProperties mqProperties;

    private final static Gson GSON = new Gson();

    public String buildDestination(String topic, String tag) {
        topic = this.reBuildTopic(topic);
        return topic + MqConstant.DESTINATION_SEPARATOR + tag;
    }

    private String reBuildTopic(String topic) {
        if (mqProperties.isEnableIsolation() && StringUtils.hasText(mqProperties.getSuffix())) {
            return topic + MqConstant.ISOLATION_SEPARATOR + mqProperties.getSuffix();
        }
        return topic;
    }

    /**
     * 发送同步消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param msg   消息体
     * @param <T>   基础message
     * @return 返回
     */
    public <T extends BaseMsg> SendResult send(String topic, String tag, T msg) {
        String destination = this.buildDestination(topic, tag);
        return this.send(destination, msg);
    }

    /**
     * 发送同步消息
     *
     * @param destination 目的
     * @param msg         消息体
     * @param <T>         基础message
     * @return 返回
     */
    public <T extends BaseMsg> SendResult send(String destination, T msg) {
        Message<T> message = MessageBuilder.withPayload(msg).setHeader(RocketMQHeaders.KEYS, msg.getKey()).build();
        SendResult result = template.syncSend(destination, message);
        log.info("[{}]同步消息[{}]发送结果[{}]", destination, GSON.toJson(message), GSON.toJson(result));
        return result;
    }

    /**
     * 发送延迟消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param msg   消息体
     * @param <T>   基础message
     * @param delay 延迟时间，单位秒
     * @return 返回
     */
    public <T extends BaseMsg> SendResult sendDelayMsg(String topic, String tag, T msg, long delay) {
        String destination = this.buildDestination(topic, tag);
        return this.sendDelayMsg(destination, msg, delay);
    }

    /**
     * 发送延迟消息
     *
     * @param destination 目的
     * @param msg         消息体
     * @param <T>         基础message
     * @param delay       延迟时间，单位秒
     * @return 返回
     */
    public <T extends BaseMsg> SendResult sendDelayMsg(String destination, T msg, long delay) {
        Message<T> message = MessageBuilder.withPayload(msg).setHeader(RocketMQHeaders.KEYS, msg.getKey()).build();
        SendResult result = template.syncSendDelayTimeSeconds(destination, message, delay);
        log.info("[{}]同步消息[{}]发送结果[{}]", destination, GSON.toJson(message), GSON.toJson(result));
        return result;
    }

    /**
     * 发送事务消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param msg   消息体
     * @param <T>   基础message
     * @return 返回
     */
    public <T extends BaseMsg> TransactionSendResult sendTxMsg(String topic, String tag, T msg) {
        String destination = this.buildDestination(topic, tag);
        return this.sendTxMsg(destination, msg);
    }

    /**
     * 发送事务消息
     *
     * @param destination 目的
     * @param msg         消息体
     * @param <T>         基础message
     * @return 返回
     */
    public <T extends BaseMsg> TransactionSendResult sendTxMsg(String destination, T msg) {
        Message<T> message = MessageBuilder.withPayload(msg).setHeader(RocketMQHeaders.KEYS, msg.getKey()).build();
        TransactionSendResult result = template.sendMessageInTransaction(destination, message, new Object());
        log.info("[{}]同步消息[{}]发送结果[{}]", destination, GSON.toJson(message), GSON.toJson(result));
        return result;
    }
}
